package net.xdclass.xdclassmq.jms.consumer;

import net.xdclass.xdclassmq.config.JmsConfig;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

//@Component
public class PayTestConsumer {

    private DefaultMQPushConsumer consumer;

    @Autowired
    private JmsConfig jmsConfig;

    public PayTestConsumer() {
    }

    @PostConstruct
    public void payTestConsumerListener() throws MQClientException {
        consumer = new DefaultMQPushConsumer(jmsConfig.getPayConsumerGroup());
        consumer.setNamesrvAddr(jmsConfig.getNameServerAddr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        //默认是集群方式，可以更改为广播，但是广播方式不支持重试
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //订阅所有标签
        consumer.subscribe(jmsConfig.getPayTopic(),"*");

        consumer.registerMessageListener((MessageListenerConcurrently)(msgs,context) -> {
            MessageExt msg = msgs.get(0);
            int times = msg.getReconsumeTimes();
            System.out.println("重复消费次数：" + times);
            try{
                System.out.printf("%s Test Receive New Messages: %s %n",
                        Thread.currentThread().getName(),
                        new String(msg.getBody(),"UTF-8"));
                String topic = msg.getTopic();
                String body = new String(msg.getBody(),"UTF-8");
                String tags = msg.getTags();
                String keys = msg.getKeys();
                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }catch (Exception e) {
                System.out.println("消费失败");
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }

        });
        consumer.start();
        System.out.println("consumer start ...");
    }


}
